package main

import (
	"database/sql"
	"fmt"
	"log"
	"strings"
	"time"

	_ "github.com/go-sql-driver/mysql"
)

// 配置结构
type Config struct {
	MySQL struct {
		Host     string `yaml:"host"`
		Port     int    `yaml:"port"`
		User     string `yaml:"user"`
		Password string `yaml:"password"`
		Database string `yaml:"database"`
		Table    string `yaml:"table"`
	} `yaml:"mysql"`

	Doris struct {
		Host     string `yaml:"host"`
		Port     int    `yaml:"port"` // Doris的MySQL协议端口，通常为9030
		User     string `yaml:"user"`
		Password string `yaml:"password"`
		Database string `yaml:"database"`
		Table    string `yaml:"table"`
	} `yaml:"doris"`

	Sync struct {
		BatchSize       int    `yaml:"batch_size"`
		Incremental     bool   `yaml:"incremental"`
		IncrementalCol  string `yaml:"incremental_col"`
		LastSyncValue   string `yaml:"last_sync_value"`
		SyncIntervalMin int    `yaml:"sync_interval_min"`
	} `yaml:"sync"`
}

func main() {
	// 初始化配置 (实际项目中可以从配置文件加载)
	config := Config{
		MySQL: struct {
			Host     string `yaml:"host"`
			Port     int    `yaml:"port"`
			User     string `yaml:"user"`
			Password string `yaml:"password"`
			Database string `yaml:"database"`
			Table    string `yaml:"table"`
		}{
			Host:     "127.0.0.1",
			Port:     3306,
			User:     "root",
			Password: "mysql_password",
			Database: "source_db",
			Table:    "tasks_scheduler",
		},
		Doris: struct {
			Host     string `yaml:"host"`
			Port     int    `yaml:"port"`
			User     string `yaml:"user"`
			Password string `yaml:"password"`
			Database string `yaml:"database"`
			Table    string `yaml:"table"`
		}{
			Host:     "127.0.0.1",
			Port:     9030, // Doris MySQL协议端口
			User:     "root",
			Password: "",
			Database: "target_db",
			Table:    "tasks_scheduler",
		},
		Sync: struct {
			BatchSize       int    `yaml:"batch_size"`
			Incremental     bool   `yaml:"incremental"`
			IncrementalCol  string `yaml:"incremental_col"`
			LastSyncValue   string `yaml:"last_sync_value"`
			SyncIntervalMin int    `yaml:"sync_interval_min"`
		}{
			BatchSize:       1000,
			Incremental:     true,
			IncrementalCol:  "update_date",
			LastSyncValue:   "2023-01-01 00:00:00",
			SyncIntervalMin: 5,
		},
	}

	// 启动同步服务
	for {
		if err := syncData(config); err != nil {
			log.Printf("同步失败: %v", err)
		} else {
			log.Println("同步完成")
		}

		// 间隔时间后再次同步
		time.Sleep(time.Duration(config.Sync.SyncIntervalMin) * time.Minute)
	}
}

// 数据同步函数
func syncData(config Config) error {
	// 1. 连接MySQL
	mysqlDB, err := connectMySQL(config.MySQL)
	if err != nil {
		return fmt.Errorf("连接MySQL失败: %v", err)
	}
	defer mysqlDB.Close()

	// 2. 连接Doris
	dorisDB, err := connectDoris(config.Doris)
	if err != nil {
		return fmt.Errorf("连接Doris失败: %v", err)
	}
	defer dorisDB.Close()

	// 3. 构建查询语句
	query := buildQuery(config)
	log.Printf("执行查询: %s", query)

	// 4. 执行MySQL查询
	rows, err := mysqlDB.Query(query)
	if err != nil {
		return fmt.Errorf("MySQL查询失败: %v", err)
	}
	defer rows.Close()

	// 5. 获取列信息
	columns, err := rows.Columns()
	if err != nil {
		return fmt.Errorf("获取列信息失败: %v", err)
	}

	// 6. 准备Doris插入语句
	insertSQL := buildInsertSQL(config.Doris.Table, columns)
	stmt, err := dorisDB.Prepare(insertSQL)
	if err != nil {
		return fmt.Errorf("准备Doris插入语句失败: %v", err)
	}
	defer stmt.Close()

	// 7. 批量处理数据
	batch := make([]interface{}, 0, config.Sync.BatchSize*len(columns))
	count := 0

	for rows.Next() {
		// 创建值容器
		values := make([]interface{}, len(columns))
		valuePtrs := make([]interface{}, len(columns))
		for i := range columns {
			valuePtrs[i] = &values[i]
		}

		// 扫描行数据
		if err := rows.Scan(valuePtrs...); err != nil {
			return fmt.Errorf("扫描行数据失败: %v", err)
		}

		// 添加到批次
		for _, v := range values {
			batch = append(batch, v)
		}
		count++

		// 达到批次大小时写入Doris
		if count%config.Sync.BatchSize == 0 {
			if err := batchInsert(stmt, batch, len(columns)); err != nil {
				return fmt.Errorf("批量插入Doris失败: %v", err)
			}
			batch = batch[:0] // 清空批次
			log.Printf("已处理 %d 条记录", count)
		}
	}

	// 写入剩余数据
	if len(batch) > 0 {
		if err := batchInsert(stmt, batch, len(columns)); err != nil {
			return fmt.Errorf("批量插入Doris失败: %v", err)
		}
		log.Printf("已处理 %d 条记录", count)
	}

	// 检查是否有行错误
	if err = rows.Err(); err != nil {
		return fmt.Errorf("行处理错误: %v", err)
	}

	return nil
}

// 连接MySQL
func connectMySQL(cfg struct {
	Host     string
	Port     int
	User     string
	Password string
	Database string
	Table    string
}) (*sql.DB, error) {
	dsn := fmt.Sprintf("%s:%s@tcp(%s:%d)/%s?charset=utf8mb4&parseTime=True&loc=Local",
		cfg.User, cfg.Password, cfg.Host, cfg.Port, cfg.Database)
	db, err := sql.Open("mysql", dsn)
	if err != nil {
		return nil, err
	}
	db.SetMaxOpenConns(10)
	db.SetMaxIdleConns(5)
	db.SetConnMaxLifetime(time.Hour)

	if err = db.Ping(); err != nil {
		return nil, err
	}
	return db, nil
}

// 连接Doris
func connectDoris(cfg struct {
	Host     string
	Port     int
	User     string
	Password string
	Database string
	Table    string
}) (*sql.DB, error) {
	dsn := fmt.Sprintf("%s:%s@tcp(%s:%d)/%s?charset=utf8mb4&parseTime=True&loc=Local",
		cfg.User, cfg.Password, cfg.Host, cfg.Port, cfg.Database)
	db, err := sql.Open("mysql", dsn) // Doris兼容MySQL协议
	if err != nil {
		return nil, err
	}
	db.SetMaxOpenConns(10)
	db.SetMaxIdleConns(5)
	db.SetConnMaxLifetime(time.Hour)

	if err = db.Ping(); err != nil {
		return nil, err
	}
	return db, nil
}

// 构建查询语句
func buildQuery(config Config) string {
	table := config.MySQL.Table
	cols := "*" // 实际项目中可以指定列

	if config.Sync.Incremental {
		return fmt.Sprintf("SELECT %s FROM %s WHERE %s > '%s' ORDER BY %s ASC",
			cols, table, config.Sync.IncrementalCol, config.Sync.LastSyncValue, config.Sync.IncrementalCol)
	}
	return fmt.Sprintf("SELECT %s FROM %s", cols, table)
}

// 构建插入语句
func buildInsertSQL(table string, columns []string) string {
	placeholders := make([]string, len(columns))
	for i := range placeholders {
		placeholders[i] = "?"
	}

	return fmt.Sprintf("INSERT INTO %s (%s) VALUES (%s)",
		table,
		strings.Join(columns, ","),
		strings.Join(placeholders, ","))
}

// 批量插入数据
func batchInsert(stmt *sql.Stmt, batch []interface{}, colCount int) error {
	if len(batch) == 0 {
		return nil
	}

	// 计算批次中的行数
	rowCount := len(batch) / colCount
	valueArgs := make([]interface{}, 0, len(batch))

	// 构建多行VALUES
	var sqlBuilder strings.Builder
	sqlBuilder.WriteString("VALUES ")

	for i := 0; i < rowCount; i++ {
		if i > 0 {
			sqlBuilder.WriteString(", ")
		}
		sqlBuilder.WriteString("(")
		for j := 0; j < colCount; j++ {
			if j > 0 {
				sqlBuilder.WriteString(", ")
			}
			sqlBuilder.WriteString("?")
			valueArgs = append(valueArgs, batch[i*colCount+j])
		}
		sqlBuilder.WriteString(")")
	}

	// 执行批量插入
	_, err := stmt.Exec(valueArgs...)
	return err
}
